import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
 * @author wsl
 * @version 2022-10-29
 */
object Receiver03_KafkaDirect {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //定义Kafka参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguiguGroup",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],

      //    "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
      //    "key.deserializer" -> classOf[StringDeserializer],
      //    "value.deserializer" -> classOf[StringDeserializer],
      //    "group.id" -> "use_a_separate_group_id_for_each_stream",

      //当自动提交偏移量时，如果各分区下有已经提交的offset时，都是从latest开始消费的；不自动提交新消费者组才从earliest开始。

      "auto.offset.reset" -> "earliest", //latest
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //读取Kafka数据创建DStream
    KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //优先位置
      ConsumerStrategies.Subscribe[String, String](Set("testTopic"), kafkaPara) // 消费策略：（订阅多个主题，配置参数）
    )
      //kafka里的消息都是<k,v>
      .map(record => record.value())
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()

    ssc.start()
    ssc.awaitTermination()
  }
}
